#include "comet_publish_request.h"

#include <set>
#include <string>
#include <sstream>
#include <iomanip>

#include <boost/lexical_cast.hpp>

#include <openssl/md5.h>
#include "base/logging.h"
#include "base/json_escape.h"
#include "uid_session_map.h"
#include "subscriber_cache.h"

#include "rpc_thread_manager.h"
#include "update_session_task.h"
#include "create_message_task.h"
#include "check_ticket_task.h"

namespace imserver {

extern boost::asio::io_service g_io_service;

int64_t CometPublishRequest::ctor_count_ = 0;
int64_t CometPublishRequest::in_200_ = 0;
int64_t CometPublishRequest::in_typing_ = 0;
int64_t CometPublishRequest::in_retry_ = 0;
int64_t CometPublishRequest::in_400_ = 0;
int64_t CometPublishRequest::in_401_ = 0;

class DuplicateFilter {
 public:
  static DuplicateFilter & Instance() {
    static DuplicateFilter filter;
    return filter;
  }

  bool Check(int64_t uid, int64_t seq) {
    if (seq == 0) {
      return true;
    }
    std::map<int64_t, int64_t>::iterator it = uid_seq_map_.find(uid);
    if (it != uid_seq_map_.end()) {
      if (it->second == seq) {
        return false;
      } else {
        it->second = seq;
      }
    } else {
      uid_seq_map_[uid] = seq;
    }
    return true;
  }
 private:
  DuplicateFilter() : reset_timer_(g_io_service), reset_interval_(3600 * 24) {
    reset_timer_.expires_from_now(boost::posix_time::seconds(reset_interval_));
    reset_timer_.async_wait(
      boost::bind(
        &DuplicateFilter::OnResetTimer,
        this,
        boost::asio::placeholders::error
      )
    );
  }

  void OnResetTimer(const boost::system::error_code& error) {
    if (error) {
      LOG_WARN("DuplicateFilter OnResetTimer error " << error.message());
    } else {
      LOG_INFO("DuplicateFilter OnResetTimer ok. uid_seq_map_.size=" << uid_seq_map_.size());
      std::map<int64_t, int64_t> empty;
      empty.swap(uid_seq_map_);
    }

    reset_timer_.expires_from_now(boost::posix_time::seconds(reset_interval_));
    reset_timer_.async_wait(
      boost::bind(
        &DuplicateFilter::OnResetTimer,
        this,
        boost::asio::placeholders::error
      )
    );
  }

  DuplicateFilter(DuplicateFilter&);
  DuplicateFilter& operator=(DuplicateFilter&);

  boost::asio::deadline_timer reset_timer_;
  int reset_interval_; // 间隔时长，单位为秒

  std::map<int64_t, int64_t> uid_seq_map_;
};

void CometPublishRequest::OnWriteComplete() {
  LOG_DEBUG("CometPublishRequest OnWriteComplete");
}

bool CometPublishRequest::Response() {
  if (CheckTicket() == 0) {
    GetThreadManager()->add(boost::shared_ptr<Runnable>(new CheckTicketTask(shared_from_this(), qid_, qid_ticket_, cookie("Q"), cookie("T"))));
  } else {
    OnCheckTicketComplete(-1, std::string());
  }
  return true;
}

int CometPublishRequest::OnCheckTicketComplete(int64_t qid, const std::string &) {
  // 注意：qid 的 0和-1 是不一样的
  if (qid >= 0) {
    qid_ = qid;
  }

  if (effective_uid() <= 0) {
    UnauthorizedResponse();
    // 掐断subscriber, 令其重连, 其必要性在于：
    // 1) 让非法(session过期等)用户重新分配session
    // 2) server重启时，client subscriber可能并未发现, 因而不会重新分配有效的session，导致消息无法正常收发
    HttpRequestPtr subscriber = SubscriberCache::Instance().Get(effective_session());
    if (subscriber) {
      subscriber->UnauthorizedResponse();
    }
    LOG_INFO("publish 401 addr=" << remote_addr());
    ++in_401_;
    return 0;
  }

  std::stringstream rsp_header;
  std::stringstream rsp_body;
  std::string callback = ParseCallback();
  if (!callback.empty()) {
    // 开头空格是 IE bug fix
    rsp_body << ' ' << callback << "(";
  }
  int64_t via_shop = 0;
  int64_t receiver = 0;
  int64_t seq = 0;
  std::string message;
  std::string msg_type;
  if (callback.empty()) {
    try {
      via_shop = boost::lexical_cast<int64_t>(post_param("vs"));
    } catch (boost::bad_lexical_cast&) {}

    try {
      receiver = boost::lexical_cast<int64_t>(post_param("receiver"));
      seq = boost::lexical_cast<int64_t>(post_param("s"));
    } catch (boost::bad_lexical_cast&) {}

    message = post_param("message");
    msg_type = post_param("type");
  } else {
    try {
      via_shop = boost::lexical_cast<int64_t>(query_param("vs"));
    } catch (boost::bad_lexical_cast&) {}

    try {
      receiver = boost::lexical_cast<int64_t>(query_param("receiver"));
      seq = boost::lexical_cast<int64_t>(post_param("s"));
    } catch (boost::bad_lexical_cast&) {}

    message = query_param("message");
    msg_type = query_param("type");
  }
  if (msg_type != "msg") {
    seq = 0;
  }

  if (receiver > 0 && (!message.empty() || msg_type == "typing")) {
    time_t timestamp = time(NULL);
    rsp_body << "{type:" << SEND_SUCCESS
       << ",sender:" << effective_uid()
       << ",receiver:" << receiver 
       << ",timestamp:" << timestamp 
       << ",message:\"" << base::JsonEscape(message)
       << "\"}";

    rsp_header << "HTTP/1.1 200 OK\r\n";
    if (DuplicateFilter::Instance().Check(effective_uid(), seq)) {
      ImMessage im_msg;
      im_msg.seq_id = 0;
      im_msg.sender = effective_uid();
      im_msg.receiver = receiver;
      im_msg.timestamp = timestamp;
      im_msg.body = message;
      if (via_shop > 0) {
        im_msg.__set_via_shop(via_shop);
      }

      if (msg_type == "typing") {
        SetMessageType(im_msg, MSG_TYPE_TYPING);
        ++in_typing_;
      } else {
        SetMessageType(im_msg, MSG_TYPE_TALK);
        ++in_200_;
      }
      GetThreadManager()->add(boost::shared_ptr<Runnable>(new CreateMessageTask(im_msg)));
     
      LOG_INFO("publish 200, sender=" << effective_uid() << " receiver="<< receiver << " message=" 
               << message << " type=" << msg_type << " addr=" << remote_addr());
    } else {
      LOG_INFO("publish retry, sender=" << effective_uid() << " receiver="<< receiver << " message=" 
               << message << " type=" << msg_type << " addr=" << remote_addr());
      ++in_retry_;
    }

    GetThreadManager()->add(boost::shared_ptr<Runnable>(new UpdateSessionTask(effective_uid(), effective_session(), true)));
  } else {
    rsp_body << "{type:" << PARAM_ERROR << ", desc:\"bad request.\"}";
    rsp_header << "HTTP/1.1 400 Bad Request\r\n";
    ++in_400_;
    LOG_INFO("publish 400, sender=" << effective_uid() << " addr=" << remote_addr());
  }

  if (!callback.empty()) {
    rsp_body << ")";
  }

  rsp_header << "Content-Type: text/javascript; charset=utf-8\r\n"
             << "Connection: close\r\n"
             << "Content-Length: " << rsp_body.str().size() << "\r\n"
             << IdCookiesHeader(qid_, qid_ticket_, temp_id_, temp_id_ticket_)
             << "\r\n";
  
  response_ = rsp_header.str();
  response_.append(rsp_body.str());

  connection_->WriteResponse(response_);
  return 0;
}

}

